package com.citusdata.elven.feeder;

import com.citusdata.elven.feeder.HttpRpcDataFeeder;
import com.citusdata.elven.linearRoad.protobuf.LinearRoad.*;
import com.citusdata.elven.linearRoad.LinearRoadCoordinator;
import com.google.protobuf.Message;
import org.apache.log4j.Logger;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ThreadPoolExecutor;
import static com.citusdata.elven.feeder.ConcurrentDataFeeder.*;


public class FileFeederRunnable implements Runnable {
  private static final Logger logger = Logger.getLogger(FileFeederRunnable.class);

  private final LinearRoadCoordinator coordinator;
  private final BufferedReader inputStream;
  private final PrintWriter    outputStream;

  private final ThreadPoolExecutor threadPool;
  private final CompletionService  completions;

  public FileFeederRunnable(LinearRoadCoordinator coordinator,
                            BufferedReader inputStream, PrintWriter outputStream) {
    this.coordinator  = coordinator;
    this.inputStream  = inputStream;
    this.outputStream = outputStream;

    this.threadPool  = HttpRpcDataFeeder.getThreadPoolExecutor();
    this.completions = new ExecutorCompletionService(threadPool); 
  }


  public void run() {
    try {
      int requestCount   = 0;
      int previousMinute = 1;
      String inputLine;

      while ((inputLine = inputStream.readLine()) != null) {
        Message request = CarDataFeeder.parseRequest(inputLine);

        int currentMinute = CarDataFeeder.getRequestSemiMinute(inputLine);
        if (currentMinute > previousMinute) {
          printQueuedResponses(completions, requestCount, outputStream);
        
          requestCount = 0;
          previousMinute = currentMinute;
        }

        requestCount++;
        completions.submit(new LinearRoadCoordinatorClient(request));
      }

      // Read the remaining responses in the queue
      printQueuedResponses(completions, requestCount, outputStream);

    } catch (IOException ex) {
      logger.error("Caught exception when feeding data", ex);
    } finally {
      threadPool.shutdown();
    }
  }

  
  // Helper class to call linear road coordinator concurrently.
  private class LinearRoadCoordinatorClient implements Callable {
    private final Message request;

    LinearRoadCoordinatorClient(Message request) {
      this.request = request;
    }

    public Message call() {
      Message response = null;

      if (request instanceof PositionRequest) {
        response = coordinator.reportPosition((PositionRequest) request);
      } else if (request instanceof AccountRequest) {
        response = coordinator.computeAccount((AccountRequest) request);
      } else if (request instanceof ExpenditureRequest) {
        response = coordinator.computeExpenditure((ExpenditureRequest) request);
      }

      return response;
    }
  }
}
